package jmind.core.dubbo.callback;

import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import jmind.core.dubbo.pojo.BusEvent;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;

public class BusServiceImpl implements IBusService {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    /** 消息队列 */
    private final BlockingQueue<BusEvent> bq = new LinkedBlockingQueue<BusEvent>();
    // 事件种类
    private final Multimap<String, Callback> listeners = HashMultimap.create();

    public BusServiceImpl() {
        System.out.println("init BusServiceImpl");
        Thread t = new Thread(new Runnable() {

            public void run() {
                while (true) {
                    try {
                        BusEvent event = bq.take();
                        Collection<Callback> collection = listeners.get(event.getTopic());
                        if (collection.size() > 0) {
                            Iterator<Callback> it = collection.iterator();
                            while (it.hasNext()) {
                                try {
                                    it.next().doIt(event);
                                } catch (com.alibaba.dubbo.rpc.RpcException e) {
                                    logger.error("doit event=", e);
                                    it.remove();
                                }
                            }
                        } else {
                            bq.add(event);
                        }
                    } catch (InterruptedException e) {
                        logger.error("", e);
                    }
                }
            }
        });
        t.setDaemon(true);
        t.start();
    }

    @Override
    public void notify(BusEvent event) {
        logger.debug("event=" + event);
        bq.offer(event);
    }

    @Override
    public void subscribe(String topic, boolean all, Callback callback) {
        logger.debug(topic + "====" + all);
        if (all || !listeners.containsKey(topic))
            listeners.put(topic, callback);

    }

    //    public void addEvent(BusEvent event) {
    //                try {
    //                    Callback callback = jmind.core.dubbo.DubboServer.getSpringBeanFactory().getBean(event.getTopic(),
    //                            Callback.class);
    //                    callback.doIt(event);
    //                } catch (Exception e) {
    //                    e.printStackTrace();
    //                }
    //
    //    }

}
